package com.gitee.dufafei.spark.connector.hbase

import com.gitee.dufafei.spark.pattern.Borrow
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase._
import org.apache.hadoop.hbase.client._

/**
 * <dependency>
 * <groupId>org.apache.hbase</groupId>
 * <artifactId>hbase-client</artifactId>
 * <version>${hbase.version}</version>
 * <scope>provided</scope>
 * </dependency>
 * <dependency>
 * <groupId>org.apache.hbase</groupId>
 * <artifactId>hbase-common</artifactId>
 * <version>${hbase.version}</version>
 * <scope>provided</scope>
 * </dependency>
 * <dependency>
 * <groupId>org.apache.hbase</groupId>
 * <artifactId>hbase-server</artifactId>
 * <version>${hbase.version}</version>
 * <scope>provided</scope>
 * </dependency>
 * <dependency>
 * <groupId>org.apache.hbase</groupId>
 * <artifactId>hbase-protocol</artifactId>
 * <version>${hbase.version}</version>
 * <scope>provided</scope>
 * </dependency>
 */
class HBaseClient(val conn: Connection) extends Borrow {

  import HBaseClient.implicits._

  def usingAdmin[A](execute:Admin => A): A = using(conn.getAdmin)(execute)

  def usingTable[A](tableName: String)(execute: Table => A): A = using(conn.getTable(tableName))(execute)

  def usingRegion[A](tableName: String)(execute: RegionLocator => A): A = using(conn.getRegionLocator(tableName))(execute)

  /**
   *  建表空间
   */
  def createNameSpace(name: String): Unit = {
    val namespaceDescriptor = NamespaceDescriptor.create(name).build()
    usingAdmin(admin => admin.createNamespace(namespaceDescriptor))
  }

  /**
   * 建表
   */
  def createTable(tableName:String, families:Array[String]): Unit = {
    val table = TableName.valueOf(tableName)
    val tableDescriptor = new HTableDescriptor(table)
    addFamilies(tableDescriptor, families)
    usingAdmin(admin => admin.createTable(tableDescriptor))
  }

  /**
   * 删表
   */
  def deleteTable(tableName: String): Unit = {
    usingAdmin{ admin =>
      admin.disableTable(tableName)
      admin.deleteTable(tableName)
    }
  }

  def addFamilies(tableDescriptor: HTableDescriptor, families:Array[String]): Unit = {
    families.foreach{ x =>
      val family = new HColumnDescriptor(x)
      tableDescriptor.addFamily(family)
    }
  }
}

/**
 * 创建Connection是重量级的,并且它是线程安全的。
 * 在整个应用（进程）的范围内只维护一个共用的Connection,比如以单例的形式,在应用退出时,再关闭连接。
 * 由Connection取得的Table和Admin对象是轻量级的,并且不是线程安全的,所以它们应该即用即弃。
 */
object HBaseClient {

  val HBASE_ZOOKERPER_QUORUM = "hbase.zookeeper.quorum"
  val HBASE_ZOOKERPER_PROPERTY_CLIENTPORT = "hbase.zookeeper.property.clientPort"
  val ZOOKERPER_ZNODE_PARENT = "zookeeper.znode.parent"

  def createConf(): Configuration = HBaseConfiguration.create()

  def createConnection(conf: Configuration): Connection = ConnectionFactory.createConnection(conf)

  implicit def cop(client: HBaseClient): HBaseCop = new HBaseCop(client)

  implicit def part(client: HBaseClient): HBasePart = new HBasePart(client)

  implicit def io(client: HBaseClient): HBaseIo = new HBaseIo(client)

  object implicits {
    implicit def convertTableName(tableName: String): TableName = TableName.valueOf(tableName)
  }
}
